//
// Created by 17775 on 2024/7/30.
//

#ifndef RAYCI_ABSTRACTQUEUEOBSERVER_H
#define RAYCI_ABSTRACTQUEUEOBSERVER_H

#include "threadsafe_linkedqueue.h"
#include "Observer.h"

template<class E, class V>
class AbstractQueueObserver : public Observer<E, V> {
protected:
    threadsafe_linkedqueue<V, std::condition_variable_any> queue;
    interruptible_thread _taskThread;


public:
    AbstractQueueObserver() : _taskThread([=]() {
        try {
            for ( ;; ) {
                if(V v; this->queue.wait_and_pop(v)) {
                    this->notifyValue(v);
                }
                else {

                }
            }
        }
        catch (...) {

        }
    }){

    }



    ~AbstractQueueObserver() override {
        if(_taskThread.joinable()) {
            _taskThread.join();
        }
    }


    void put(V&& value) override {
        queue.push(std::forward<V>(value));
    }

    void put(const V& value) override {
        queue.push(value);
    }

    void interrupt() {
        _taskThread.interrupt();
    }

};


#endif //RAYCI_ABSTRACTQUEUEOBSERVER_H
